package cn.dfun.sample.flink.apitest

import org.apache.flink.api.common.functions.ReduceFunction
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.{EventTimeSessionWindows, SlidingProcessingTimeWindows, TumblingEventTimeWindows}
import org.apache.flink.streaming.api.windowing.time.Time

// date +%s
object WindowTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    // 事件时间语义
    // 默认时间语义为processing time
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
//    env.getConfig.setAutoWatermarkInterval(50)
//    val inputPath = "C:\\wor\\flink-sample\\src\\main\\resources\\sensor"
//    val inputStream= env.readTextFile(inputPath)
    val inputStream = env.socketTextStream("node-01", 7777)

    // 包装成样例类
    val dataStream = inputStream
      .map(data => {
        var arr = data.split(",")
        SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
      })
      // 从当前数据提取时间戳并指定watermark生成机制
      .assignTimestampsAndWatermarks(
        new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(3)) {
          override def extractTimestamp(t: SensorReading): Long = {
            t.timestamp * 1000L
          }
        })
//      .assignAscendingTimestamps(_.timestamp * 1000L) // 升序数据时间戳,毫秒
    val lateTag = new OutputTag[(String, Double, Long)]("late")
    // 每15秒统计一次窗口内所有数据最小值(滚动窗口),以及最新的时间戳
    val resultStream = dataStream
        .map(data => (data.id, data.temperature, data.timestamp))
        .keyBy(_._1)
        .timeWindow(Time.seconds(15))
        .allowedLateness(Time.minutes(1))
        .sideOutputLateData(lateTag)
//        .minBy(1)
        .reduce((curResult, newData) => (curResult._1, curResult._2.min(newData._2), newData._3))
    // 滚动时间窗口
//      .window(TumblingEventTimeWindows.of(Time.seconds(15)))
    // 滑动时间窗口
//      .window(SlidingProcessingTimeWindows.of(Time.seconds(15), Time.seconds(3)))
    // 会话窗口
//        .window(EventTimeSessionWindows.withGap(Time.seconds(10)))
      // 简写 滚动1参数 滑动2参数
//      .timeWindow(Time.seconds(15))
//        .countWindow(10)
    resultStream.getSideOutput(lateTag).print("late")
    resultStream.print()
    env.execute("window test")
  }
}

class MyReducer extends ReduceFunction[SensorReading] {
  override def reduce(t: SensorReading, t1: SensorReading): SensorReading = {
    SensorReading(t.id, t.timestamp, t.temperature.min(t1.timestamp))
  }
}
